package cn.xiaoniu.dmp.report

import cn.xiaoniu.dmp.helper.ConfigHelper
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext}

object AreaDistribute {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("统计省市地域分布")
    val sc = new SparkContext(conf)
    val ssc = new SQLContext(sc)

    val frame: DataFrame = ssc.read.parquet("D:/dmp/parquet")
    frame.registerTempTable("t_basic")

    val result: DataFrame = ssc.sql(
      """
      select provincename,cityname,
      sum(case when requestmode=1 and processnode>=1 then 1 else 0 end)originalReq,
      sum(case when requestmode=1 and processnode>=2 then 1 else 0 end)effectiveReq,
      sum(case when requestmode=1 and processnode>=3 then 1 else 0 end)adReq,

      sum(case when iseffective=1 and isbilling=1 and isbilling=1 and adorderid !=0 then 1 else 0 end)bid,
      sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1 else 0 end)bidsucc,

      sum(case when requestmode=2 and iseffective=1 then 1 else 0 end)display,
      sum(case when requestmode=3 and iseffective=1 then 1 else 0 end)click,

      sum(case when iseffective=1 and isbilling=1 and iswin=1 then winprice/1000 else 0 end)consume,
      sum(case when iseffective=1 and isbilling=1 and iswin=1 then adpayment/1000 else 0 end)cost
       from t_basic
      group by provincename,cityname
      """.stripMargin)
    result.write.jdbc(ConfigHelper.url, ConfigHelper.areatable, ConfigHelper.props)

    sc.stop()
  }

}
